package com.xiaotao.realtimeui.spark


import com.fasterxml.jackson.databind.ObjectMapper
import com.xiaotao.realtimeui.po.PaymentInfo
import com.xiaotao.realtimeui.utils.RedisUtils
import org.apache.kafka.clients.consumer.ConsumerConfig._
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}
import org.slf4j.LoggerFactory
import org.springframework.stereotype.Component


@Component
class StreamingProcessor {
  def process(): Unit = {

    val conf: SparkConf = new SparkConf().setAppName("KafkaStreamingTest").setMaster("local[2]")
    val sc = new  SparkContext(conf)
    val log = LoggerFactory.getLogger(this.getClass)
    val orderTotalKey = "bussiness::order::total::"
    val totalKey = "bussiness:order:all"
    sc.setLogLevel("WARN")
    val ssc = new StreamingContext(sc, Seconds(1))
    val kafkaParams = Map(
      BOOTSTRAP_SERVERS_CONFIG -> "master:9092",
      GROUP_ID_CONFIG -> "spark-receiver",
      KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer]
    )
    val topics = Set("itcast_order")
    val mapper = new ObjectMapper();
    KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    ).map(record => {
      val info = mapper.readValue(record.value(), classOf[PaymentInfo])
      (info.getProductId, info.getProductPrice)
    }).groupByKey().map(x => (x._1, x._2.size, x._2.sum))
    .foreachRDD(rdds => {
      rdds.foreachPartition(infos => {
        infos.foreach(info => {
          log.info("读到数据，处理中")
          RedisUtils.increment(orderTotalKey + info._1, info._3)
          RedisUtils.increment(totalKey, info._3)
        })
      })
    })

    ssc.start()
    ssc.awaitTermination()
  }
}
